from __future__ import annotations
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Any, Callable, Optional, Sequence, Union
from core.thread.protocol import Readable
from core.logger.proxy import logger

# ---------- 工具 ----------


def _convert_list(obj: Union[Any, Sequence[Any], None]) -> list[Any]:
    if obj is None:
        return []
    if isinstance(obj, (list, tuple)):
        return list(obj)
    return [obj]


# ---------- 消费者线程 ----------


class QueueConsumerThread(threading.Thread):
    """
    通用消费线程：
    - source: 实现了 read() 的对象；可选实现 ack()/nack()
    - handler: 业务处理函数 (item -> None 或返回值)
    - 并发：内部线程池对批次内 item 并发处理
    """

    def __init__(
        self,
        *,
        source: Readable,
        handler: Callable[[Any], Any],
        name: Optional[str] = None,
        stop_event: Optional[threading.Event] = None,
        max_workers: int = 8,
        batch_size: int = 1,
        read_timeout: float = 2.0,
        backoff_base: float = 0.2,
        backoff_max: float = 2.0,
    ):
        super().__init__(name=name, daemon=True)
        self.source = source
        self.handler = handler
        self.pool = ThreadPoolExecutor(
            max_workers=max_workers, thread_name_prefix=f"{name}-qct-pool"
        )
        self.stop_event = stop_event or threading.Event()
        self.batch_size = batch_size
        self.read_timeout = read_timeout
        self.backoff_base = backoff_base
        self.backoff_max = backoff_max

    def run(self):
        logger.info(f"consumer {self.name} <started>...")
        backoff = self.backoff_base
        try:
            while not self.stop_event.is_set():
                try:
                    items = _convert_list(
                        self.source.read(self.batch_size, self.read_timeout)
                    )
                    if not items:
                        # 空轮询：指数退避，避免空转
                        time.sleep(backoff)
                        backoff = min(self.backoff_max, backoff * 2)
                        continue

                    backoff = self.backoff_base  # 有数据就复位退避

                    futures = {self.pool.submit(self._handle, i) for i in items}
                    for fut in as_completed(futures):
                        # 触发异常抛出，便于观测；个别失败不影响其他
                        _ = fut.result()

                except Exception as e:
                    logger.error(f"consumer {self.name} loop error: %s", e)
                    time.sleep(self.backoff_max)

        finally:
            self.pool.shutdown(wait=True)
            logger.info(f"consumer {self.name} <stopped>.")

    def _handle(self, item: Any):
        try:
            self.handler(item)
        except Exception as e:
            logger.error(f"consumer {self.name} handle failed: %s", e)
